package com.bw.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.base.BaseApp;
import com.bw.gmall.realtime.common.bean.TrafficPageViewBean;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.function.DorisMapFunction;
import com.bw.gmall.realtime.common.util.DateFormatUtil;
import com.bw.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.table.runtime.functions.KeyedProcessFunctionWithCleanupState;
import org.apache.flink.util.Collector;
import org.apache.hadoop.crypto.key.JavaKeyStoreProvider;

import java.time.Duration;
import java.util.Iterator;

public class DwsTrafficVcChArIsNewPageViewWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsTrafficVcChArIsNewPageViewWindow().start(Constant.TOPIC_DWD_TRAFFIC_PAGE,Constant.DWS_TRAFFIC_VC_CH_AR_IS_NEW_PAGE_VIEW_WINDOW,4,10022);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {

//        dataStreamSource.print();

        // 1. 解析数据, 封装 pojo 中  uv sv pv during_time

        SingleOutputStreamOperator<JSONObject> streamOperator = etlStream(dataStreamSource);
//        streamOperator.print();


       //分组
        KeyedStream<JSONObject, String> keyBy = streamOperator.keyBy(x -> x.getJSONObject("common").getString("mid"));


        //计算 uv pv sv
        SingleOutputStreamOperator<TrafficPageViewBean> streamOperator1 = uvpv_svStream(keyBy);

        //水位线
        SingleOutputStreamOperator<TrafficPageViewBean> waterStream = waterStream(streamOperator1);


        // 2. 开窗聚合 vc ch ar is_new
        KeyedStream<TrafficPageViewBean, Tuple4<String, String, String, String>> keyedStream = keyOther(waterStream);

        //reduce 计算
        SingleOutputStreamOperator<TrafficPageViewBean> streamOperator2 = reduceStream(keyedStream);

//        streamOperator2.print();
        // 3. 写出到 doris 中
        streamOperator2.map(new DorisMapFunction<>()).sinkTo(FlinkSinkUtil.getDorisSink(Constant.DORIS_DATABASE+".dws_traffic_vc_ch_ar_is_new_page_view_window",Constant.DWS_TRAFFIC_VC_CH_AR_IS_NEW_PAGE_VIEW_WINDOW));



    }

    private static SingleOutputStreamOperator<TrafficPageViewBean> reduceStream(KeyedStream<TrafficPageViewBean, Tuple4<String, String, String, String>> keyedStream) {
        return keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(1)))
                .reduce(new ReduceFunction<TrafficPageViewBean>() {
                    @Override
                    public TrafficPageViewBean reduce(TrafficPageViewBean t1, TrafficPageViewBean t2) throws Exception {
                        //滚动没有重复数据 所以不需要new reduce 原理就是 2位数 进行计算，第一位数 用来统计本次结果，第二个数用来接收新来的数据

                        t1.setSvCt(t1.getSvCt() + t2.getSvCt());
                        t1.setPvCt(t1.getPvCt() + t2.getPvCt());
                        t1.setUvCt(t1.getUvCt() + t2.getUvCt());
                        t1.setDurSum(t1.getDurSum() + t2.getDurSum());

                        return t1;
                    }
                }, new WindowFunction<TrafficPageViewBean, TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple4<String, String, String, String> s, TimeWindow timeWindow, Iterable<TrafficPageViewBean> iterable, Collector<TrafficPageViewBean> collector) throws Exception {

                        //获取对应的 时间
                        long start = timeWindow.getStart();
                        long end = timeWindow.getEnd();

                        //获取数据 进行 存值
                        Iterator<TrafficPageViewBean> iterator = iterable.iterator();

                        TrafficPageViewBean trafficPageViewBean = iterator.next();

                        trafficPageViewBean.setStt(DateFormatUtil.tsToDate(start));
                        trafficPageViewBean.setEdt(DateFormatUtil.tsToDate(end));

                        // 今天
                        trafficPageViewBean.setCur_date(DateFormatUtil.tsToDate(System.currentTimeMillis()));

                        collector.collect(trafficPageViewBean);
                    }
                });
    }

    private static KeyedStream<TrafficPageViewBean, Tuple4<String, String, String, String>> keyOther(SingleOutputStreamOperator<TrafficPageViewBean> waterStream) {
        return waterStream.keyBy(new KeySelector<TrafficPageViewBean, Tuple4<String, String, String, String>>() {
            @Override
            public Tuple4<String, String, String, String> getKey(TrafficPageViewBean trafficPageViewBean) throws Exception {
                return new Tuple4<>(trafficPageViewBean.getVc(), trafficPageViewBean.getCh(), trafficPageViewBean.getAr(), trafficPageViewBean.getIsNew());
            }
        });
    }

    private static SingleOutputStreamOperator<TrafficPageViewBean> waterStream(SingleOutputStreamOperator<TrafficPageViewBean> streamOperator1) {
        return streamOperator1.assignTimestampsAndWatermarks(WatermarkStrategy
                .<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                .withTimestampAssigner((event, timestamp) -> event.getTs()));
    }

    private static SingleOutputStreamOperator<TrafficPageViewBean> uvpv_svStream(KeyedStream<JSONObject, String> keyBy) {
        return keyBy.process(new ProcessFunction<JSONObject, TrafficPageViewBean>() {
            ValueState<String> lastDtState;

            @Override
            public void open(Configuration parameters) {
                ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("state", String.class);

                stateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(org.apache.flink.api.common.time.Time.hours(24))
                        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite).build()
                );


                lastDtState = getRuntimeContext().getState(stateDescriptor);
            }

            @Override
            public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, TrafficPageViewBean>.Context context, Collector<TrafficPageViewBean> collector) throws Exception {
                Long ts = jsonObject.getLong("ts");

                String curDt = DateFormatUtil.tsToDate(ts);
                JSONObject common = jsonObject.getJSONObject("common");
                JSONObject page = jsonObject.getJSONObject("page");
                String lastValue = lastDtState.value();

                Long uvCt = 0L;
                Long svCt = 0L;
                if (!curDt.equals(lastValue)) {   //注意不可将 有空值可能的放入 前面
                    // 独立
                    uvCt = 1L;
                    lastDtState.update(curDt);
                }

                String last_page_id = page.getString("last_page_id");
                if (last_page_id == null) {
                    // 会话数
                    svCt = 1L;
                }


                TrafficPageViewBean trafficPageViewBean = new TrafficPageViewBean();

                trafficPageViewBean.setAr(common.getString("ar"));
                trafficPageViewBean.setIsNew(common.getString("is_new"));
                trafficPageViewBean.setCh(common.getString("ch"));
                trafficPageViewBean.setTs(ts);
                trafficPageViewBean.setDurSum(page.getLong("during_time"));
                trafficPageViewBean.setSid(common.getString("sid"));
                trafficPageViewBean.setUvCt(uvCt);
                trafficPageViewBean.setPvCt(1L);
                trafficPageViewBean.setVc(common.getString("vc"));
                trafficPageViewBean.setSvCt(svCt);

                collector.collect(trafficPageViewBean);
            }
        });
    }


    private static SingleOutputStreamOperator<JSONObject> etlStream(DataStreamSource<String> dataStreamSource) {
        return dataStreamSource.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String s, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {

                if (s != null) {

                    JSONObject jsonObject = JSON.parseObject(s);

                    if (jsonObject != null) {
                        JSONObject common = jsonObject.getJSONObject("common");
                        Long ts = jsonObject.getLong("ts");
                        if (common != null && ts != null) {
                            String mid = common.getString("mid");

                            if (mid != null) {
                                collector.collect(jsonObject);
                            }
                        }

                    }
                }
            }
        });
    }

}
